package com.example.demo20250926.provider;

import com.alibaba.fastjson.JSONObject;
import com.example.demo20250926.model.Order;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.time.LocalDateTime;

/**
 * @author sky
 * @date 2025年10月31日 14:21
 * @description
 * @since 1.0.0
 */
@Component
@Slf4j
public class KafkaProvider{
    /**
     * 消息 TOPIC
     */
    private static final String TOPIC = "topic-demo";
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    /**
     * 发送消息
     *
     * @param message
     */
    public void sendMessage(long orderId, String orderNum, LocalDateTime createTime) {
        log.info("### 发送消息：{}", orderNum);
        // 构建一个订单类
        Order order = Order.builder().orderId(orderId).orderNum(orderNum).orderTime(createTime).build() ;
        // 发送消息，订单类的 json 作为消息体
        ListenableFuture<SendResult<String,String>> future = kafkaTemplate.send(TOPIC, JSONObject.toJSONString( order));
        // 监听回调
        future.addCallback(new ListenableFutureCallback<SendResult<String,String>>(){
            @Override
            public void onSuccess(SendResult<String,String> result) {
                log.info("### 发送成功：{}", result);
            }
            @Override
            public void onFailure(Throwable ex) {
                log.error("### 发送失败：{}", ex.getMessage());
            }

        });
    }
}
